package top.pcat.service;

import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;
import top.pcat.config.KafkaConfig;
import top.pcat.config.ServerConfig;
import top.pcat.entity.ANSI;
import top.pcat.entity.ChatRoom;
import top.pcat.entity.Message;
import top.pcat.entity.Sender;
import top.pcat.util.KafkaProducerUtil;
import top.pcat.util.ProtostuffUtils;
import top.pcat.util.RedPacket;
import top.pcat.util.RedisPoolUtil;

import java.util.*;

import static top.pcat.util.RedisPoolUtil.getJedis;

@Slf4j
public class MessagingService {

    /**
     * 处理收到的 Kafka 消息
     * @param bytes 数据
     */
    public void processMsgReceivedByKafka(byte[] bytes) {
        Message msg = ProtostuffUtils.deserialize(bytes, Message.class);
        log.info("收到kafka消息---> value = " + msg);
        // 判断是否是本机发送的消息
        if (msg.getServerId()!=null) {
            if (!msg.getServerId().equals(ServerConfig.serverId)) {
                // 随机消息
                if (msg.getController() == 5) {

                    log.info("收到随机消息");
                    log.info("检查senderToChannelMap==================================");
                    for (Map.Entry<String, Channel> entry : ServerConfig.senderToChannelMap.entrySet()) {
                        log.info(entry.getKey());
                        log.info(entry.getValue().toString());
                    }
                    log.info("完成senderToChannelMap==================================");

                    String[] split = msg.getMesType().split(":"); //要发送给的用户的id
                    log.info("随机消息要发给" + Arrays.toString(split));

                    // 遍历数组找到要发送的人
                    for (Channel c : ServerConfig.channelGroup) {
                        if (!ServerConfig.senderToChannelMap.isEmpty() && ServerConfig.senderToChannelMap.get(split[0]) == c) {
                            log.info("找到要发送随机消息的用户了");
                            c.writeAndFlush(msg);
                        }
                    }

                } else {
                    log.info("是普通kafka消息 ---> value = " + msg);
                    long roomId = msg.getRoomId();
                    log.info("检查channelToRoomIdMap数据" + ServerConfig.channelToRoomIdMap);
                    // 遍历数组找到属于这个聊天室的人
                    for (Channel c : ServerConfig.channelGroup) {
                        if (!ServerConfig.channelToRoomIdMap.isEmpty() && ServerConfig.channelToRoomIdMap.get(c) == roomId)
                            c.writeAndFlush(msg);
                    }
                }
            }
        }else{
            log.info("接收到的kafka消息异常 getServerId null");
        }
    }

    /**
     * 发送普通消息
     * @param message
     */
    public void normalMessage(Message message) {

        //发给本服务器的
        for(Channel c : ServerConfig.channelGroup){
            System.out.println(ServerConfig.channelToRoomIdMap);
            if (ServerConfig.channelToRoomIdMap.get(c)==message.getRoomId()) {
                c.writeAndFlush(message);
            }
        }

        // 发给kafka
        log.info("通过kafka发送消息"+message);
        byte[] serializeMessage = ProtostuffUtils.serialize(message);
        new KafkaProducerUtil(new KafkaConfig()).sendMsgByKafka("test",serializeMessage);

    }

    /**
     * 向客户端发送的第一条消息,让他选择聊天室
     * @param message
     */
    public void firstMessage(Message message, Channel channel) {

        String senderId = message.getSenderId();
        String senderName = message.getSenderName();

        //添加在线用户列表
        Map<String,String> senderinfo = new HashMap<>();
        senderinfo.put(String.valueOf(senderId),senderName);
        Jedis jedis = getJedis();
        long hset = jedis.hset("user:1001", senderinfo);

        log.info("注册用户上线 ----->  "+message.getSenderId()+":"+message.getSenderName()+":"+hset);
        ServerConfig.senderToChannelMap.put(senderId,channel);

        roomInfoMessage(message,channel);
    }

    public void roomInfoMessage(Message message, Channel channel){
        String senderId = message.getSenderId();
        String senderName = message.getSenderName();
        log.info("通过redis获取聊天室列表");
        List<ChatRoom> roomList = new ChatRoomService().getRoomList();
        String ls = "[";
        for (ChatRoom c : roomList){
            ls = ls + c.getRoomId() + "." + c.getRoomName() +" ";
        }
        ls += "] 请输入序号，新建聊天室请输入：& <聊天室名称>";

        Message res = new Message("sys", "sys", 0, "null", "chooseChatRoom", " 选择聊天室 " + ls, 1, new Date());

        channel.writeAndFlush(res);

        ServerConfig.channelToSenderMap.put(channel,new Sender(senderId,senderName));
    }

    /**
     * 发送新成员消息
     * @param msg 消息
     * @param channel channel
     */
    public void newMemberMessage(Message msg, Channel channel) {

        log.info("收到选择聊天室消息");
        ServerConfig.channelToRoomIdMap.put(channel,Long.valueOf(msg.getMesContent()));

        log.info("放入map");
        String senderId = msg.getSenderId();
        String senderName = msg.getSenderName();
        long roomId = msg.getRoomId();
        ServerConfig.channelToSenderMap.get(channel).setRoomId(roomId);

        log.info("redis用户加入聊天室"+roomId+" : "+senderId+" : "+ServerConfig.serverId);
        new ChatRoomService().addUserToChatroom(roomId,senderId,ServerConfig.serverId);

        log.info("广播消息"+senderName+" 加入聊天室");
        msg.setMesContent(senderName+" 加入聊天室");
        msg.setController(1);
        normalMessage(msg);
    }

    /**
     * 发送红包
     * @param msg 消息
     * @param channel channel
     */
    public void redPacketMessage(Message msg, Channel channel) {

        log.info("收到红包消息");

        String mesContent1 = msg.getMesContent();
        String[] split = mesContent1.split("-");
        log.info("红包个数："+split[0]+"金额"+split[1]);

        RedPacket redPacket = new RedPacket();
        redPacket.setAmount(Integer.parseInt(split[1])*100);
        redPacket.setTotal(Integer.valueOf(split[0]));
        redPacket.setUserId(msg.getSenderId());

        log.info("收到就绪");
        //发红包
        RedPacketService redPacketService = new RedPacketService();
        String res = redPacketService.newRedPacket(redPacket);
        if (res.equals("金额或数量异常")){
            log.info("红包金额或数量异常");
            Message message = new Message(msg.getSenderId(),msg.getSenderName(),msg.getRoomId(),msg.getRoomName(),msg.getMesType(),"红包金额或数量异常",1,new Date());
            channel.writeAndFlush(message);
        }else{
            String times = res.substring(3,8);

            log.info("广播有红包消息");

            String s = "\n"+ ANSI.ANSI_ORIGIN+ANSI.ANSI_RED+"　　　　　　　　　　　　　　　　　　"+ANSI.ANSI_RESET+"\n"+
                    ANSI.ANSI_ORIGIN+ANSI.ANSI_RED+"　　"+ANSI.ANSI_RED_BACKGROUND+ANSI.ANSI_YELLOW+ANSI.ANSI_BOLD+"　∪　"+ANSI.ANSI_RESET+ANSI.ANSI_ORIGIN+ANSI.ANSI_BOLD+ANSI.ANSI_BLACK+"　　大吉大利，晚上吃鸡　 　"+ANSI.ANSI_RESET+"\n"+
                    ANSI.ANSI_ORIGIN+ANSI.ANSI_RED+"　　"+ANSI.ANSI_RED_BACKGROUND+"　 　"+ANSI.ANSI_RESET+ANSI.ANSI_ORIGIN+ANSI.ANSI_BOLD+ANSI.ANSI_BLACK+"　　发送％"+times+"抢红包　 　"+ANSI.ANSI_RESET+"\n"+
                    ANSI.ANSI_ORIGIN+ANSI.ANSI_RED+"　　　　　　　　　　　　　　　　　　"+ANSI.ANSI_RESET;

            Message message = new Message(msg.getSenderId(),msg.getSenderName(),msg.getRoomId(),msg.getRoomName(),msg.getMesType(),s,2,new Date());
            message.setServerId(ServerConfig.serverId);
            normalMessage(message);
        }

    }

    /**
     * 抢红包结果消息
     * @param msg 消息
     * @param channel channel
     */
    public void redPacketResultMessage(Message msg, Channel channel) {

        log.info("收到抢红包消息");

        String hbtimes = msg.getMesContent();
        JedisPool jedisPool = RedisPoolUtil.getRedisPoolUtil();
        Jedis jedis = jedisPool.getResource();
        log.info("搜索"+"red:packet:???"+hbtimes+"*");
        Set<String> keys = jedis.keys("red:packet:???*"+hbtimes+"*");
        log.info("找到的redis红包"+keys);

        if (keys.isEmpty()){
            log.info("红包列表为空");

            Message messagehb = new Message();
            messagehb.setRoomId(msg.getRoomId());
            messagehb.setRoomName(msg.getRoomName());
            messagehb.setController(3);
            messagehb.setMesContent("红包已过期");
            messagehb.setDate(new Date());
            messagehb.setSenderId("1");
            messagehb.setSenderName("系统");
            messagehb.setMesType("qianghongbao");
            messagehb.setDate(new Date());
            messagehb.setServerId(ServerConfig.serverId);
            channel.writeAndFlush(messagehb);

        }else{
            String firstItem = null;
            firstItem = new ArrayList<>(keys).get(0);
            RedPacketService redPacketService1 = new RedPacketService();

            log.info("找到的redis红包"+firstItem);
            int b = redPacketService1.grabRedPacket(firstItem);

            if (b!=0){
                Message messagehb = new Message();
                messagehb.setRoomId(msg.getRoomId());
                messagehb.setRoomName(msg.getRoomName());
                messagehb.setController(3);
                messagehb.setMesContent("抢到红包啦~ 金额是："+ ((double) b) /10 + "元");
                messagehb.setDate(new Date());
                messagehb.setSenderId("1");
                messagehb.setSenderName("系统");
                messagehb.setMesType("qianghongbao");
                messagehb.setDate(new Date());
                messagehb.setServerId(ServerConfig.serverId);
                channel.writeAndFlush(messagehb);
            }else{
                Message messagehb = new Message();
                messagehb.setRoomId(msg.getRoomId());
                messagehb.setRoomName(msg.getRoomName());
                messagehb.setController(3);
                messagehb.setMesContent("没抢到红包");
                messagehb.setDate(new Date());
                messagehb.setSenderId("1");
                messagehb.setSenderName("系统");
                messagehb.setMesType("qianghongbao");
                messagehb.setDate(new Date());
                messagehb.setServerId(ServerConfig.serverId);
                channel.writeAndFlush(messagehb);
            }

        }
    }

    /**
     * 发送随机消息
     * @param msg 消息
     * @param channel channel
     */
    public void randomMessage(Message msg, Channel channel) {

        //随机选择聊天室人员

        String randomMembers = ChatRoomService.getRandomChatroomMembers(msg.getRoomId());
        log.info("找到的人："+randomMembers);

        //kafka 发送消息

        Message newMessage = new Message();
        newMessage.setRoomId(msg.getRoomId());
        newMessage.setRoomName(msg.getRoomName());
        newMessage.setController(5);
        newMessage.setMesContent(msg.getMesContent()+" (随机消息)");
        newMessage.setDate(new Date());
        newMessage.setSenderId(msg.getSenderId());
        newMessage.setSenderName(msg.getSenderName());
        newMessage.setMesType(randomMembers);
        newMessage.setDate(new Date());
        newMessage.setServerId(ServerConfig.serverId);

        log.info("判断是否是本机的随机消息"+channel.toString());
        assert randomMembers != null;
        String[] split1 = randomMembers.split(":");

        //发给本服务器的
        for(Channel c : ServerConfig.channelGroup){
            if (ServerConfig.senderToChannelMap.get(split1[0])==c) {
                log.info("是本机的随机消息，netty发送");
                c.writeAndFlush(newMessage);
                return ;
            }
        }

        log.info("向kafka发送随机消息"+newMessage);
        byte[] serializeMessage = ProtostuffUtils.serialize(newMessage);
        new KafkaProducerUtil(new KafkaConfig()).sendMsgByKafka("test",serializeMessage);


        //给发送者一个反馈
        Jedis jedis = getJedis();
        String hget = jedis.hget("user:1001", split1[0]);
        log.info("要随机发送的用户名"+hget);
        Message res = new Message("sys", "sys", msg.getRoomId(), msg.getRoomName(), "random rec", "已将随机消息发送至用户: ["+hget+"] ", 1, new Date());

        channel.writeAndFlush(res);
    }

    /**
     * 建立新的聊天室
     * @param msg
     * @param channel
     */
    public void newRoomMessage(Message msg, Channel channel) {

        Jedis jedis = getJedis();
        String roomName = msg.getRoomName();
        String maxIdKey = "roomMaxId";
        String roomsHash = "chatRoom";
        String existingRooms = "chatRoomName";

        while (true) {
            // 监视max_id键和existing_rooms键
            jedis.watch(maxIdKey, existingRooms);
            // 判断房间名是否已存在
            if (jedis.sismember(existingRooms, roomName)) {
                log.info("新建聊天室失败，聊天室已存在");
                String newMsg = "新建聊天室失败，聊天室已存在，"+msg.getMesContent();
                msg.setMesContent(newMsg);
                roomInfoMessage(msg,channel);
                break; // 房间名已存在，退出循环
            }
            // 获取当前最大ID值
            String maxIdStr = jedis.get(maxIdKey);
            int maxId = maxIdStr == null ? 0 : Integer.parseInt(maxIdStr);
            // 开始事务
            Transaction tx = jedis.multi();
            // 自增ID值并将新的ID和房间名保存到Redis的hash中
            tx.hset(roomsHash, String.valueOf(++maxId), roomName);
            // 更新最大ID值
            tx.set(maxIdKey, String.valueOf(maxId));
            // 将新的房间名保存到SET中
            tx.sadd(existingRooms, roomName);
            // 提交事务
            List<Object> result = tx.exec();
            if (result == null) {
                // 如果事务执行失败，说明max_id或existing_rooms键在监视期间被其他客户端修改了，需要重试
                continue;
            }
            log.info("新建聊天室成功");
            String newMsg = "新建聊天室成功，"+msg.getMesContent();
            msg.setMesContent(newMsg);
            roomInfoMessage(msg,channel);
            break; // 事务执行成功，退出循环
        }

    }
}
